package com.bindatax.core.task.segment;

import com.bindatax.core.async.callback.ICallback;
import com.bindatax.core.async.callback.IWorker;
import com.bindatax.core.async.worker.WorkResult;
import com.bindatax.core.async.wrapper.WorkerWrapper;
import com.bindatax.core.meta.TaskMeta;
import com.bindatax.core.task.DataTask;
import com.bindatax.core.dspool.DataTaskConnManager;
import com.bindatax.core.log.LogOutInterface;
import com.bindatax.core.log.LogOutManager;
import com.bindatax.core.statistics.ExecuteStatistics;
import com.bindatax.core.task.event.EventType;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class SegmentWork implements SegmentInterface
        , IWorker<SegmentData, SegmentData>, ICallback<SegmentData, SegmentData> {
    private final static LogOutInterface LOG = LogOutManager.get();

    private final SegmentData segmentData;
    private volatile boolean notifyErrorStop = false;
    private volatile boolean notifyDone = false;

    private Thread thisThread;

    private final ExecuteStatistics tempStatistics = new ExecuteStatistics();
    public SegmentWork(SegmentData segmentData) {
        this.segmentData = segmentData;
        this.segmentData.status = SegmentStatus.INIT;
    }

    @Override
    public void begin() {
        this.segmentData.status = SegmentStatus.READY;
    }

    @Override
    public SegmentData action(SegmentData object, Map<String, WorkerWrapper> allWrappers) {
        if (notifyErrorStop) {
            this.segmentData.status = SegmentStatus.FINISH;
            return segmentData;
        }
        thisThread = Thread.currentThread();
        this.segmentData.status = SegmentStatus.EXECUTION;
        Connection targetConn = null;
        PreparedStatement targetStatement = null;
        DataTask parent = segmentData.parent;
        TaskMeta taskMeta = parent.getTaskMeta();

        int batchSize = taskMeta.getWriteSegmentBatchSize();
        List<Object[]> batchBuffRowsList = new ArrayList<>(batchSize);

        try {
            targetConn = DataTaskConnManager.getConnection(segmentData.connKey);
            targetConn.setAutoCommit(false);
            targetStatement = targetConn.prepareStatement(segmentData.sql);
            int tryNum = taskMeta.getFailTryNum();
            int stopTimeInSeconds = taskMeta.getFailTryInterval() * 1000;

            while (true) {
                //如果外部通知从源库获取数据完成，且数据队列为空，则停止入库。
                if(this.notifyDone && segmentData.allSegmentDataQueue.size() == 0){
                    LOG.debug("检测到任务组通知【数据入库完成】，且公共数据池已无数据，退出循环。");
                    break;
                }
                //从数据队列中获取数据，最多等1秒钟。
                Object[] sourceColRows = segmentData.allSegmentDataQueue.poll(1, TimeUnit.SECONDS);
                if (sourceColRows != null) {
                    batchBuffRowsList.add(sourceColRows);
                }
                //查看是否满了
                if (batchBuffRowsList.size() >= batchSize) {
                    long batchId = System.currentTimeMillis();
                    LOG.debug("批次"+batchId + " - 准备入库批量数据...." );
                    segmentData.parent.exeEvent(this, EventType.TASK_RUNNING_SEGMENT_BEFORE);
                    parent.executeBatch(targetConn, targetStatement, batchBuffRowsList, tryNum, stopTimeInSeconds, tempStatistics , batchId , this);
                    parent.addStatistics( this, tempStatistics );
                    tempStatistics.clear();
                    LOG.debug("批次"+batchId + " - 完成批量入库：" + batchBuffRowsList.size() +"条");
                    batchBuffRowsList.clear();
                }
                //外部命令停止，就停止当前进程。
                if (this.notifyErrorStop) {
                    LOG.debug("检测到任务组通知【停止数据入库】，退出循环获取公共数据池数据。");
                    segmentData.parent.exeEvent(this, EventType.TASK_RUNNING_SEGMENT_NOTIFY_ERROR_STOP);
                    break;
                }
            }

            int remainSize = batchBuffRowsList.size();
            // 得看看临时数据列表还有没有剩余的数据
            if (this.notifyDone) {
                if( remainSize > 0){
                    // 如果是通知数据完成的，需要把当前的数据入库完成
                    long batchId = System.currentTimeMillis();
                    LOG.debug("批次"+batchId + " - 准备入库剩余片内数据...." );
                    parent.executeBatch(targetConn, targetStatement, batchBuffRowsList, tryNum, stopTimeInSeconds, tempStatistics,batchId , this);
                    parent.addStatistics( this, tempStatistics);
                    tempStatistics.clear();
                    LOG.debug("批次"+batchId + " - 完成入库剩余片内数据：" +remainSize+"条，数据片线程已停止。");
                    batchBuffRowsList.clear();
                }else {
                    LOG.debug("已无剩余数据，数据片线程已停止。");
                }
                segmentData.parent.exeEvent(this, EventType.TASK_RUNNING_SEGMENT_AFTER_SUCCESS);
            }
            if(this.notifyErrorStop){
                if(remainSize > 0){
                    LOG.debug("当前数据片因其他数据片错误而停止，不再处理片内数据，把数据放回公共数据池：" + remainSize + "条");
                    for (int i = 0; i < remainSize; i++) {
                        Object[] objects = batchBuffRowsList.get(i);
                        segmentData.allSegmentDataQueue.offer(objects,1000,TimeUnit.MILLISECONDS);
                    }
                    batchBuffRowsList.clear();
                }
            }
        } catch (Exception e) {
            segmentData.errorException = e;
            LOG.error("分片入库线程执行失败！", e);
            // 通知其他线程停止
            this.segmentData.parent.stopAllSegmentErrorStop();
            segmentData.parent.exeEvent(this, EventType.TASK_RUNNING_SEGMENT_AFTER_FAIL);
        } finally {
            thisThread = null;
            parent.addStatistics(this,tempStatistics );
            tempStatistics.clear();
            this.segmentData.status = SegmentStatus.FINISH;
            DataTaskConnManager.close(null, targetStatement, targetConn);
        }
        return segmentData;
    }

    @Override
    public void result(boolean success, SegmentData param, WorkResult<SegmentData> workResult) {
        this.segmentData.status = SegmentStatus.FINISH;
    }

    @Override
    public SegmentData defaultValue() {
        return this.segmentData;
    }

    @Override
    public SegmentData getSegmentData() {
        return this.segmentData;
    }

    @Override
    public void notifyErrorStop() {
        if(this.segmentData.status == SegmentStatus.FINISH){
            return;
        }
        this.notifyErrorStop = true;
    }

    @Override
    public void notifyDone() {
        this.notifyDone = true;
    }

    @Override
    public void interruptSegment() {
        if(thisThread != null){
            LOG.warn("尝试中断数据片线程：" + thisThread.getName());
            thisThread.interrupt();
            thisThread = null;
        }
    }
}
